package com.atguigu.edu.realtime.utils;


import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;

import java.io.IOException;

//kafka工具类
public class MyKafkaUtil {

    //部署到服务器运行地址
    private static final String KAFKA_SERVER = "hadoop102:9092,hadoop103:9092,hadoop104:9092,hadoop105:9092,hadoop106:9092";

    //获取kafkasource
    public static KafkaSource<String> getKafkaSource (String topic , String groupId){

        return KafkaSource.<String>builder()
                .setBootstrapServers(KAFKA_SERVER)
                .setTopics(topic)
                .setGroupId(groupId)
                .setValueOnlyDeserializer(
                        new DeserializationSchema<String>() {
                            @Override
                            public String deserialize(byte[] message) throws IOException {
                                if (message != null){
                                    return new String(message);
                                }
                                return null;
                            }

                            @Override
                            public boolean isEndOfStream(String nextElement) {
                                return false;
                            }

                            @Override
                            public TypeInformation<String> getProducedType() {
                                return TypeInformation.of(String.class);
                            }
                        }
                )
                .setStartingOffsets(OffsetsInitializer.latest())
                //.setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed")
                .build();
    }


    public static KafkaSink<String> getKafkaSink(String topic) {
        KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
                .setBootstrapServers(KAFKA_SERVER)
                .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                        .setTopic(topic)
                        .setValueSerializationSchema(new SimpleStringSchema())
                        .build()
                )
                //在生产环境下，要想保证写入到kafka数据的一致性，必须做如下操作
                //开启检查点
                //设置DeliveryGuarantee.EXACTLY_ONCE
                // .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                //设置事务前缀
                // .setTransactionalIdPrefix("xxx")
                //在消费端，设置事务的读取隔离级别为读已提交 // .setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed")
                //设置事务的超时时间
                // .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 15*60*1000 + "")
                .build();
        return kafkaSink;
    }

    //获取从topic_db主题中读取数据  创建动态表的建表语句
    public static String getTopicDbDDL(String groupId) {
        return "CREATE TABLE topic_db (\n" +
                "`database` string,\n" +
                "`table` string,\n" +
                "`type` string,\n" +
                "ts string ,\n" +
                "`xid` string ,\n" +
                "`commit` boolean ,\n" +
                "`data` map<string,string> ,\n" +
                "`old` map<string,string>,\n" +
                "proc_time as PROCTIME()\n"+
                ") " + getKafkaDDL("topic_db", groupId);
    }

    //获取kafka连接器连接属性
    public static String getKafkaDDL(String topic, String groupId) {
        return " WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = '" + topic + "',\n" +
                "  'properties.bootstrap.servers' = '" + KAFKA_SERVER + "',\n" +
                "  'properties.group.id' = '" + groupId + "',\n" +
                //在生产环境下，为了保证消费的一致性，需要做如下的配置
                //"  'scan.startup.mode' = 'group-offsets',\n" +
                //"  'properties.auto.offset.reset' = 'latest',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'format' = 'json'\n" +
                ")";
    }

    //获取upsert-kafka连接器的连接属性
    public static String getUpsertKafkaDDL(String topic) {
        return " WITH (\n" +
                "  'connector' = 'upsert-kafka',\n" +
                "  'topic' = '" + topic + "',\n" +
                "  'properties.bootstrap.servers' = '" + KAFKA_SERVER + "',\n" +
                "  'key.format' = 'json',\n" +
                "  'value.format' = 'json'\n" +
                ")";
    }
}
